package io.confluent.kafka.consumer.processor;

import io.confluent.kafka.consumer.conf.Attributes;
import io.confluent.kafka.consumer.conf.PropertiesReader;
import io.confluent.kafka.consumer.entities.EventAlarmBean;
import io.confluent.kafka.consumer.entities.event.EventModel;
import io.confluent.util.HttpUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.codehaus.jackson.annotate.JsonProperty;
import org.codehaus.jackson.map.ObjectMapper;
import java.io.IOException;
import java.text.DecimalFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Pattern;

/**
 * Created by taotao on 2017/12/19.
 */
public class EventProcessor {
    public static void postData(ConsumerRecords<String, String> records) {
        Long count = new Long(0);
        List<EventAlarmBean> alarms = new ArrayList<EventAlarmBean>();
        if ("true".equals(PropertiesReader.DEBUG) && records.count() > 0) {
            System.out.println("拉取数目：" + records.count());
        }
        for (ConsumerRecord<String, String> record : records) {
            alarms.add(EventProcessor.parseObject(record.value()));
            count++;
            if (count % 50 == 0 || count == records.count()) {
                String jsons = EventProcessor.convertJson(alarms);
                if ("true".equals(PropertiesReader.DEBUG)) {
                    System.out.println("UMC Json数据：" + jsons);
                }
                if (!"".equals(jsons)) {
                    HttpUtil.doPost(Attributes.REQURL, jsons);
                }
                //插入完成后清空缓存
                alarms.clear();
            }
        }

    }

    public static String convertJson(List<EventAlarmBean> alarms) {
        String body = "";
        try {
            body = new ObjectMapper().writeValueAsString(alarms);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return body;
    }

    public static EventAlarmBean parseObject(String json) {
        EventAlarmBean alarm = new EventAlarmBean();
        ObjectMapper mapper = new ObjectMapper();
        EventModel eventModel = null;
        try {
            eventModel = mapper.readValue(json, EventModel.class);
            //保留原始告警消息
            alarm = generateObject(eventModel, json);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return alarm;
    }

    private static EventAlarmBean generateObject(EventModel eventModel, String json) {
        EventAlarmBean eventAlarmBean = new EventAlarmBean();
        Map<String, String> attributes = getAttributes(eventModel);

        int alarmType = getAlarmType(attributes.get(Attributes.EVENTTYPE));
        if (alarmType != 0) { //处理已定义事件，其它未知事件直接返回
            Date dateNow =  new Date();
            eventAlarmBean.setAlarmKey(attributes.get(Attributes.RESOURCE_ID) + "-" + String.valueOf(System.currentTimeMillis()) + "-" + UUID.randomUUID());
            eventAlarmBean.setMoc(Attributes.MOC_EVENT);
            eventAlarmBean.setResourceId(attributes.get(Attributes.RESOURCE_ID));
            eventAlarmBean.setTenantId(attributes.get(Attributes.TENANT_ID));
            eventAlarmBean.setHost(attributes.get(Attributes.HOST));
            eventAlarmBean.setUserId(attributes.get(Attributes.USER_ID));
            eventAlarmBean.setAlarmName(eventModel.getAlarmName());
            eventAlarmBean.setPerceivedSeverity(getSeverity(eventModel.getSeverity()));
            eventAlarmBean.setCurrentStatus(eventModel.getCurrent());
            eventAlarmBean.setPreviousStatus(eventModel.getPrevious());
            eventAlarmBean.setAlarmRuleId(eventModel.getAlarmId());
            eventAlarmBean.setReason(eventModel.getReason());
            eventAlarmBean.setEventType(eventModel.getReasonData().getType());
            eventAlarmBean.setAlarmRaisedTime(eventModel.getReasonData().getEvent().getGenerated());
            eventAlarmBean.setAlarmRaisedTime_gmt(eventModel.getReasonData().getEvent().getGenerated());
            eventAlarmBean.setReceivedTime(dateNow);
            eventAlarmBean.setOriginaldata(json);
        }

        return eventAlarmBean;
    }

    private static Map<String, String> getAttributes(EventModel eventModel) {
        Map<String, String> attributes = new HashMap<String, String>();
        attributes.put(Attributes.EVENTTYPE, eventModel.getReasonData().getEvent().getEventType());
        attributes.put(Attributes.MESSAGE_ID, eventModel.getReasonData().getEvent().getEventType());
        for (List<String> items: eventModel.getReasonData().getEvent().getTraits()) {
            if (Attributes.RESOURCE_ID.equals(items.get(0))) {
                attributes.put(Attributes.RESOURCE_ID, items.get(2));
            } else if (Attributes.USER_ID.equals(items.get(0))) {
                attributes.put(Attributes.USER_ID, items.get(2));
            } else if (Attributes.TENANT_ID.equals(items.get(0))) {
                attributes.put(Attributes.TENANT_ID, items.get(2));
            } else if (Attributes.HOST.equals(items.get(0))) {
                attributes.put(Attributes.HOST, items.get(2));
            }
        }
        return attributes;
    }

    /**
     * 判断告警类型。注意:Aodh Event针对定义的每个Openstack操作都有两条事件，.start事件和.end事件。
     * 程序只处理.end结尾的事件，.start事件不处理，因为.start事件中不包括必要的字段域，例如resource_id等。
     * @param eventType
     * @return
     */
    private static int getAlarmType(String eventType) {
        int code = 0;
        String pattern1 = "compute.instance.*.end";
        String pattern2 = "volume.*.end";
        String pattern3 = "snapshot.*.end";
        String pattern4 = "image.*.end";
        String pattern5 = "orchestration.stack.*.end";
        String pattern6 = "sahara.cluster.*.end";
        String pattern7 = "identity.*.end";
        String pattern8 = "objectstore.http.request.end";
        String pattern9 = "network.*.end";
        String pattern10 = "subnet.*.end";
        String pattern11 = "port.*.end";
        String pattern12 = "router.*.end";
        String pattern13 = "floatingip.*.end";
        String pattern14 = "pool.*.end";
        String pattern15 = "vip.*.end";
        String pattern16 = "member.*.end";
        String pattern17 = "health_monitor.*.end";
        String pattern18 = "healthmonitor.*.end";
        String pattern19 = "listener.*.end";
        String pattern20 = "loadbalancer.*.end";
        String pattern21 = "firewall.*.end";
        String pattern22 = "firewall_policy.*.end";
        String pattern23 = "firewall_rule.*.end";
        String pattern24 = "vpnservice.*.end";
        String pattern25 = "ipsecpolicy.*.end";
        String pattern26 = "ikepolicy.*.end";
        String pattern27 = "ipsec_site_connection.*.end";
        String pattern28 = "\\*http.*.end";
        String pattern29 = "dns.domain.*.end";
        String pattern30 = "trove.*.end";
        String pattern31 = "profiler.*.end";
        String pattern32 = "magnum.bay.*.end";
        String pattern33 = "compute.metrics.*.end";
        if (Pattern.matches(pattern1, eventType)) code = 7;
        else if (Pattern.matches(pattern2, eventType)) code = 8;
        else if (Pattern.matches(pattern3, eventType)) code = 9;
        else if (Pattern.matches(pattern4, eventType)) code = 10;
        else if (Pattern.matches(pattern5, eventType)) code = 11;
        else if (Pattern.matches(pattern6, eventType)) code = 12;
        else if (Pattern.matches(pattern7, eventType)) code = 13;
        else if (Pattern.matches(pattern8, eventType)) code = 14;
        else if (Pattern.matches(pattern9, eventType)) code = 15;
        else if (Pattern.matches(pattern10, eventType)) code = 16;
        else if (Pattern.matches(pattern11, eventType)) code = 17;
        else if (Pattern.matches(pattern12, eventType)) code = 18;
        else if (Pattern.matches(pattern13, eventType)) code = 10;
        else if (Pattern.matches(pattern14, eventType)) code = 20;
        else if (Pattern.matches(pattern15, eventType)) code = 21;
        else if (Pattern.matches(pattern16, eventType)) code = 22;
        else if (Pattern.matches(pattern17, eventType)) code = 23;
        else if (Pattern.matches(pattern18, eventType)) code = 24;
        else if (Pattern.matches(pattern19, eventType)) code = 25;
        else if (Pattern.matches(pattern20, eventType)) code = 26;
        else if (Pattern.matches(pattern21, eventType)) code = 27;
        else if (Pattern.matches(pattern22, eventType)) code = 28;
        else if (Pattern.matches(pattern23, eventType)) code = 29;
        else if (Pattern.matches(pattern24, eventType)) code = 30;
        else if (Pattern.matches(pattern25, eventType)) code = 31;
        else if (Pattern.matches(pattern26, eventType)) code = 32;
        else if (Pattern.matches(pattern27, eventType)) code = 33;
        else if (Pattern.matches(pattern28, eventType)) code = 34;
        else if (Pattern.matches(pattern29, eventType)) code = 35;
        else if (Pattern.matches(pattern30, eventType)) code = 36;
        else if (Pattern.matches(pattern31, eventType)) code = 37;
        else if (Pattern.matches(pattern32, eventType)) code = 38;
        else if (Pattern.matches(pattern33, eventType)) code = 39;
        else  code = 0;
        return code;
    }

    private static long getUUID()
    {
        Random r = new Random();
        //如生成的随机位数不足6位则自动加零补充
        DecimalFormat g = new DecimalFormat("10000");
        //返回时间增量+随机数的序列
        return Long.parseLong(String.format("%s%s", System.currentTimeMillis(), g.format(r.nextInt(10000))));
    }

    private static int getSeverity(String severity) {
        int severityCode = 0;
        if (Attributes.SEVERITY_LOW.equals(severity)) severityCode = 3;
        else if (Attributes.SEVERITY_MODERATE.equals(severity)) severityCode = 2;
        else if (Attributes.SEVERITY_CRITICAL.equals(severity)) severityCode = 1;
        //其它类型设成不确定告警
        else severityCode = 5;
        return severityCode;
    }

    public static Date str2Date(String dateStr) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("", Locale.SIMPLIFIED_CHINESE);
        dateFormat.applyPattern("yyyy-MM-dd'T'HH:mm:ss");
        Date date = null;
        try {
            date = dateFormat.parse(dateStr);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return date;
    }

}
